import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.{Dataset, SparkSession}

/**
 * 2、日志分析
 * 日志格式：IP、命中率(Hit/Miss)、响应时间、请求时间、请求方法、请求URL、请求协议、状态码、响应大小、referer、用户代理
 * 日志文件位置：data/cdn.txt
 * 术语解释：
 * PV(page view)，即页面浏览量；衡量网站或单一网页的指标
 * uv(unique visitor)，指访问某个站点或点击某条新闻的不同IP地址的人数
 * 任务：
 * 2.1、计算独立IP数
 * 2.2、统计每个视频独立IP数（视频的标志：在日志文件的某些可以找到 *.mp4，代表一个视频文件）
 * 2.3、统计一天中每个小时的流量
 */

object Subject2 {
  def main(args: Array[String]): Unit = {

    Logger.getLogger("org").setLevel(Level.WARN)
    val spark = SparkSession
      .builder()
      .appName(this.getClass.getCanonicalName)
      .master("local[*]")
      .getOrCreate()
    import spark.implicits._
    import spark._

    //读cdn.txt，获得ip地址、访问时间（小时）、请求的url、状态码、响应大小
    val dsLog: Dataset[(String, String, String, String, String)] = spark.read
      .option("delimiter", " ")
      .csv("data/cdn.txt")
      .map { row =>
        (row.getString(0), row.getString(3).split(":")(1),
          row.getString(5).split("\\s+")(1), row.getString(6), row.getString(7))
      }

    dsLog.show(10)

    //第一列是ip地址，去重计数
    println("独立IP数：" + dsLog.map(_._1).distinct.count)
    println("**********************************************************")

    val dsVideo: Dataset[((String, String), Int)] = dsLog.map { line =>
      //第三列是请求的url，用/做分隔符得到请求的文件名
      val str: Array[String] = line._3.split("/")
      val fileName: String = str(str.length - 1)
      //判断文件是否是视频文件(.mp4后缀)
      if (fileName.endsWith(".mp4")) {
        //((视频文件名, ip地址),1)
        ((fileName, line._1), 1)
      } else
        (null, 0)
    }

    dsVideo.show
    println("**********************************************************")

    //过滤掉为空的数据
    dsVideo.where('_1.isNotNull)
      //以第一列(fileName,ip)为key，统计对于某个视频，同一个ip的访问次数
      .groupBy('_1)
      .count()
      //(fileName, 1)
      .map{ row => (row.getStruct(0).getString(0), 1)}
      //以视频名fileName为key，统计对于某个视频，一共有多少个独立ip访问过
      .groupBy('_1)
      .count().show()
    println("**********************************************************")

    dsLog.map{ line =>
      //第四列是状态码，获取正常返回的请求(状态码为200或206或304)
      val status: String = line._4
      status match {
        //(小时, 响应大小)
        case "200" | "206" | "304" => (line._2, line._5.toLong)
        case _ => (null, 0L)
      }
    }
      //过滤空值
      .where('_1.isNotNull)
      //以第一列(小时)为key，统计每个小时的流量
      .groupBy('_1)
      .sum()
      .sort('_1).show(24)

  }

}
